package data

import (
	"encoding/json"
	"fmt"
	"log"
	"sync"
)

// DataService 数据服务的主入口，提供统一的数据处理接口
type DataService struct {
	dataProcessor   *DataProcessor
	kafkaConsumer   *KafkaConsumer
	mongodbHandler  *MongoDBHandler
	handlers        map[string]func(string, string, []byte) error
	handlersLock    sync.RWMutex
	isRunning       bool
}

// DataServiceConfig 数据服务的配置参数
type DataServiceConfig struct {
	KafkaBrokers  []string
	MongoURI      string
	MongoDBName   string
	MongoCollection string
}

// NewDataService 创建一个新的数据服务实例
func NewDataService(config DataServiceConfig) (*DataService, error) {
	// 初始化数据处理器
	dataProcessor, err := NewDataProcessor(
		config.KafkaBrokers,
		config.MongoURI,
		config.MongoDBName,
		config.MongoCollection,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create data processor: %v", err)
	}

	// 初始化MongoDB处理器
	mongodbHandler, err := NewMongoDBHandler(
		config.MongoURI,
		config.MongoDBName,
		config.MongoCollection,
	)
	if err != nil {
		dataProcessor.Close()
		return nil, fmt.Errorf("failed to create mongodb handler: %v", err)
	}

	// 创建索引
	if err := mongodbHandler.CreateIndexes(); err != nil {
		log.Printf("Warning: failed to create indexes: %v", err)
	}

	// 初始化Kafka消费者
	kafkaConsumer, err := NewKafkaConsumer(config.KafkaBrokers)
	if err != nil {
		mongodbHandler.Close()
		dataProcessor.Close()
		return nil, fmt.Errorf("failed to create kafka consumer: %v", err)
	}

	service := &DataService{
		dataProcessor:  dataProcessor,
		kafkaConsumer:  kafkaConsumer,
		mongodbHandler: mongodbHandler,
		handlers:       make(map[string]func(string, string, []byte) error),
	}

	return service, nil
}

// Start 启动数据服务
func (ds *DataService) Start() error {
	if ds.isRunning {
		return nil
	}

	// 启动Kafka消费者，消费用于模型训练的数据
	err := ds.kafkaConsumer.ConsumeTopic("iot-device-data", ModelTrainingHandler)
	if err != nil {
		return fmt.Errorf("failed to start kafka consumer: %v", err)
	}

	ds.isRunning = true
	log.Println("Data service started successfully")
	return nil
}

// Stop 停止数据服务
func (ds *DataService) Stop() error {
	if !ds.isRunning {
		return nil
	}

	// 停止Kafka消费者
	ds.kafkaConsumer.Stop()

	// 关闭数据处理器
	err := ds.dataProcessor.Close()
	if err != nil {
		log.Printf("Warning: failed to close data processor: %v", err)
	}

	// 关闭MongoDB处理器
	err = ds.mongodbHandler.Close()
	if err != nil {
		log.Printf("Warning: failed to close mongodb handler: %v", err)
	}

	ds.isRunning = false
	log.Println("Data service stopped")
	return nil
}

// HandleDeviceData 处理从设备接收到的数据
func (ds *DataService) HandleDeviceData(deviceID string, topic string, payload []byte) error {
	// 解析payload数据
	var data map[string]interface{}
	if err := json.Unmarshal(payload, &data); err != nil {
		return fmt.Errorf("failed to parse device data: %v", err)
	}

	// 处理数据
	err := ds.dataProcessor.ProcessData(deviceID, topic, data)
	if err != nil {
		return fmt.Errorf("failed to process device data: %v", err)
	}

	// 触发注册的处理函数
	ds.triggerHandlers(deviceID, topic, payload)

	return nil
}

// RegisterHandler 注册数据处理函数
func (ds *DataService) RegisterHandler(name string, handler func(string, string, []byte) error) {
	ds.handlersLock.Lock()
	ds.handlers[name] = handler
	ds.handlersLock.Unlock()

	log.Printf("Registered data handler: %s", name)
}

// UnregisterHandler 取消注册数据处理函数
func (ds *DataService) UnregisterHandler(name string) {
	ds.handlersLock.Lock()
	delete(ds.handlers, name)
	ds.handlersLock.Unlock()

	log.Printf("Unregistered data handler: %s", name)
}

// triggerHandlers 触发所有注册的处理函数
func (ds *DataService) triggerHandlers(deviceID string, topic string, payload []byte) {
	ds.handlersLock.RLock()
	handlers := make(map[string]func(string, string, []byte) error, len(ds.handlers))
	for name, handler := range ds.handlers {
		handlers[name] = handler
	}
	ds.handlersLock.RUnlock()

	for name, handler := range handlers {
		go func(n string, h func(string, string, []byte) error) {
			err := h(deviceID, topic, payload)
			if err != nil {
				log.Printf("Error in data handler %s: %v", n, err)
			}
		}(name, handler)
	}
}

// GetDeviceData 获取设备的历史数据
func (ds *DataService) GetDeviceData(deviceID string, limit int64) ([]map[string]interface{}, error) {
	results, err := ds.mongodbHandler.GetDeviceData(deviceID, limit)
	if err != nil {
		return nil, fmt.Errorf("failed to get device data: %v", err)
	}

	// 转换为map[string]interface{}
	var data []map[string]interface{}
	for _, result := range results {
		item := make(map[string]interface{})
		for k, v := range result {
			item[k] = v
		}
		data = append(data, item)
	}

	return data, nil
}

// IsRunning 检查数据服务是否正在运行
func (ds *DataService) IsRunning() bool {
	return ds.isRunning
}